#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "msg_queue_peer.h"
#include <sqlite3.h>
#include <pthread.h>
#include <time.h>

#define CLIENTID "ExampleClientPub"
#define TOPIC "MQTT Examples"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
#define shmemwhere "/tmp/ipc/shmem/%s"
#define shmemname "shmem"
#define msgqueuewhere "/tmp/ipc/msgqueue/peer/"
#define msgqueuename "msgqueue"
#define pointtablewhere "/mnt/config/pointtable.json"
#define sqlite3dbwhere "/mnt/hgfs/class/val.db"

//消息队列 共享内存  数据库 点表
sqlite3 *db;
char *errmsg;
union val_t {

    int bool;
    int a;
    float b;
};

struct std_node
{
    int key;
    int type;
    int type_dev;
    union val_t old_val;
    union val_t new_val;
    int ret;
};

struct msgqueue
{
    int typekey;
    char data[64];
};

MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;

volatile MQTTClient_deliveryToken deliveredtoken;
void publicmqtt(char *p)
{
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;

    MQTTClient_publishMessage(client, "up", &pubmsg, &token);
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    payloadptr = message->payload;
    //接收mqtt
    printf("%s\n", payloadptr);

    cJSON *root = cJSON_Parse(payloadptr);
    cJSON *item = cJSON_GetObjectItem(root, "type");
    int type = item->valueint;

    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    if (type == 1)
    {
        //单次返回数据
        //序列化
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        cJSON *data = cJSON_CreateArray();
        cJSON_AddItemToObject(root, "data", data);
        cJSON *item;
        for (int i = 0; i < *p; i++)
        {
            item = cJSON_CreateObject();
            cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
            char val[64] = {0};
            if (dev->type == 1)
            {
                sprintf(val, "%d", dev->new_val.bool);
            }
            else if (dev->type == 2)
            {
                sprintf(val, "%d", dev->new_val.a);
            }
            else if (dev->type == 3)
            {
                sprintf(val, "%f", dev->new_val.b);
            }
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
            cJSON_AddItemToArray(data, item);
            dev++;
        }

        publicmqtt(cJSON_Print(root));
    }
    else if (type == 2) //控制功能
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "key");
        int key = item->valueint;
        item = cJSON_GetObjectItem(data, "val");
        char val[64] = {0};
        strcpy(val, item->valuestring);

        //写到消息队列
        struct msgqueue recvque;
        recvque.typekey = key;
        printf("%s\n", val);
        strcpy(recvque.data, val);
        msg_queue_send(msgqueuename, &recvque, sizeof(struct msgqueue), 0);
        // struct msgqueue writeque;
        // msg_queue_recv("msgqueue", &writeque, sizeof(struct msgqueue), 101, 0);
        // printf("%d  %s\n",writeque.typekey,writeque.data);
    }
    else if (type == 3) //模式切换
    {

        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "type");
        int type2 = item->valueint;

        int fd = open(pointtablewhere, O_RDWR);
        char *buf1 = (char *)malloc(4096);
        int len = read(fd, buf1, 4096);
        lseek(fd, 0, SEEK_SET);
        cJSON *root1 = cJSON_Parse(buf1);
        cJSON *report = cJSON_GetObjectItem(root1, "report");

        if (type2 == 0) //主动采集
        {
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(0));
            write(fd, cJSON_Print(root1), strlen(cJSON_Print(root1)));
            //手动重启进程
        }
        else if (type2 == 1) //变化上报
        {
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(1));
            write(fd, cJSON_Print(root1), strlen(cJSON_Print(root1)));
        }
        else if (type2 == 2) //周期上报
        {
            item = cJSON_GetObjectItem(data, "period");
            int period = item->valueint;
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(2));
            cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(period));
            write(fd, cJSON_Print(root1), strlen(cJSON_Print(root1)));
        }
        close(fd);
    }
    else if (type == 4) //历史查询
    {

        char *errmsg;
        char **resultp;
        int n_row, n_cloum;
        int i, j;

        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "key");
        int key = item->valueint;
        printf("查询%d历史\n", key);
        cJSON *limit_array = cJSON_GetObjectItem(data, "limit");
        long limitmin = cJSON_GetArrayItem(limit_array, 0)->valueint;
        long limitmax = cJSON_GetArrayItem(limit_array, 1)->valueint;
        //查
        char sql_cmd[128] = {0};
        sprintf(sql_cmd, "select time,val from history where key=%d and time>=%ld and time <=%ld;", key, limitmin, limitmax);

        if (sqlite3_get_table(db, sql_cmd, &resultp, &n_row, &n_cloum, &errmsg) != SQLITE_OK)
        {
            printf("err:%s\n", errmsg);
            exit(-1);
        }
        char **p;
        p = resultp + n_cloum;
        //序列化
        cJSON *root1 = cJSON_CreateObject();
        cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(4));
        cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
        cJSON_AddItemToObject(root1, "key", cJSON_CreateNumber(key));
        cJSON *data1 = cJSON_CreateArray();
        cJSON_AddItemToObject(root1, "data", data1);
        cJSON *item1;
        printf("%d %d\n", n_row, n_cloum);
        for (i = 0; i < n_row; i++)
        {
            item1 = cJSON_CreateObject();
            for (j = 0; j < n_cloum; j++)
            {
                if (j == 0)
                {
                    double time;
                    sscanf(p[(i * n_cloum) + j], "%lf", &time);
                    cJSON_AddItemToObject(item1, "time", cJSON_CreateNumber(time));
                    printf("time:%s\n", p[(i * n_cloum) + j]);
                }
                if (j == 1)
                {
                    cJSON_AddItemToObject(item1, "val", cJSON_CreateString(p[(i * n_cloum) + j]));
                    printf("val:%s\n", p[(i * n_cloum) + j]);
                }
                // printf("值:%s\n",p[(i * n_cloum) + j]);
            }
            cJSON_AddItemToArray(data1, item1);
        }
        printf("%s\n", cJSON_Print(root1));
        publicmqtt(cJSON_Print(root1));
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

void *handler_thread(void *arg)
{
    char sql_cmd[128] = {0};
    time_t current_time;

    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    char val[64] = {0};
    while (1)
    {
        sleep(5);
        dev = dev1;
        for (int i = 0; i < *p; i++)
        {
            time(&current_time);
            if (dev->type == 1)
            {
                sprintf(val, "%d", dev->new_val.bool);
            }
            else if (dev->type == 2)
            {

                sprintf(val, "%d", dev->new_val.a);
            }
            else if (dev->type == 3)
            {
                sprintf(val, "%f", dev->new_val.b);
            }
            sprintf(sql_cmd, "INSERT INTO history VALUES (%d, %s, %ld );", dev->key, val, current_time);
            sqlite3_exec(db, sql_cmd, NULL, NULL, &errmsg);
            dev++;
        }
    }
    return NULL;
}

int main(int argc, char *argv[])
{
    printf("上报进程启动\n");

    //打开数据库写

    char *errmsg;
    //打开数据库
    int ret = sqlite3_open(sqlite3dbwhere, &db);
    if (ret != SQLITE_OK)
    {
        printf("err open:%s\n", sqlite3_errmsg(db));
        return -1;
    }

    //读点表确定上报方式

    FILE *fp;
    fp = fopen(pointtablewhere, "r");

    char *buf1 = (char *)malloc(4096);

    fread(buf1, 4096, 1, fp);

    cJSON *root = cJSON_Parse(buf1);
    cJSON *report = cJSON_GetObjectItem(root, "report");

    cJSON *item = cJSON_GetObjectItem(report, "type");
    int type = item->valueint;

    //mqtt
    char mqttaddr[64] = {0};
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    sprintf(mqttaddr, "tcp://%s:%d", cJSON_GetObjectItem(mqtt_server, "addr")->valuestring, cJSON_GetObjectItem(mqtt_server, "port")->valueint);

    MQTTClient_create(&client, mqttaddr, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    char buf[32] = {0};
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, "down", QOS);

    //开辟共享内存，写key值
    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *data = cJSON_GetObjectItem(stm32, "data");
    int stm32num = cJSON_GetArraySize(data);

    for (int i = 0; i < stm32num; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dev->key = key->valueint;
        dev->type = type->valueint;
        dev->type_dev = 0;
        dev->ret = -1;
        dev++;
    }
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    data = cJSON_GetObjectItem(modbus, "data");
    int modbusnum = cJSON_GetArraySize(data);
    for (int i = 0; i < modbusnum; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dev->key = key->valueint;
        dev->type = type->valueint;
        dev->type_dev = 1;
        dev->ret = -1;
        dev++;
    }
    *p = stm32num + modbusnum;
    dev = dev1;

    //开线程写历史记录
    pthread_t tid;
    if (pthread_create(&tid, NULL, handler_thread, NULL) != 0)
    {
        printf("error\n");
    }

    if (type == 0)
    {
        printf("刷新上报\n");
    }
    else if (type == 1) //变化
    {
        printf("先上报一次\n");
        //先上报一次
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        cJSON *data = cJSON_CreateArray();
        cJSON_AddItemToObject(root, "data", data);
        cJSON *item;
        for (int i = 0; i < *p; i++)
        {
            item = cJSON_CreateObject();
            cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
            char val[64] = {0};
            if (dev->type == 1)
            {
                sprintf(val, "%d", dev->new_val.bool);
            }
            else if (dev->type == 2)
            {
                sprintf(val, "%d", dev->new_val.a);
            }
            else if (dev->type == 3)
            {
                sprintf(val, "%f", dev->new_val.b);
            }
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
            cJSON_AddItemToArray(data, item);
            dev++;
        }

        publicmqtt(cJSON_Print(root));

        while (1)
        {
            sleep(2);
            //序列化
            dev = dev1;
            cJSON *root = cJSON_CreateObject();
            cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
            cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
            cJSON *data = cJSON_CreateArray();
            cJSON_AddItemToObject(root, "data", data);
            cJSON *item;
            // for (int i = 0; i < 8; i++)
            // {
            //     printf("%d\n",dev->key);
            //     dev++;
            // }
            dev = dev1;
            for (int i = 0; i < *p; i++)
            {
                // printf("%d\n",*p);
                item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
                char val[64] = {0};

                //  printf("%d**%d\n",dev->new_val.bool,dev->old_val.bool);
                //判断新旧值是否相等
                if (dev->type == 1)
                {
                    if (dev->new_val.bool == dev->old_val.bool)
                    {
                        dev++;
                        continue;
                    }
                    else
                    {
                        sprintf(val, "%d", dev->new_val.bool);
                    }
                }
                else if (dev->type == 2)
                {
                    if (dev->new_val.a == dev->old_val.a)
                    {
                        dev++;
                        continue;
                    }
                    else
                    {
                        sprintf(val, "%d", dev->new_val.a);
                    }
                }
                else if (dev->type == 3)
                {
                    if (dev->new_val.b == dev->old_val.b)
                    {
                        dev++;
                        continue;
                    }
                    else
                    {
                        sprintf(val, "%f", dev->new_val.b);
                    }
                }
                cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                cJSON_AddItemToArray(data, item);
                dev++;
            }
            item = cJSON_GetArrayItem(data, 0);
            if (item == NULL)
            {
                printf("无变化\n");
                continue;
            }
            publicmqtt(cJSON_Print(root));
        }
    }
    else if (type == 2)
    {
        //周期
        item = cJSON_GetObjectItem(report, "period");
        int period = item->valueint;
        //共享内存读
        while (1)
        {
            sleep(period);
            //序列化
            dev = dev1;
            cJSON *root1 = cJSON_CreateObject();
            cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(1));
            cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
            cJSON *data1 = cJSON_CreateArray();
            cJSON_AddItemToObject(root1, "data", data1);
            cJSON *item1;
            for (int i = 0; i < *p; i++)
            {
                item1 = cJSON_CreateObject();
                cJSON_AddItemToObject(item1, "key", cJSON_CreateNumber(dev->key));
                char val[64] = {0};
                if (dev->type == 1)
                {
                    sprintf(val, "%d", dev->new_val.bool);
                }
                else if (dev->type == 2)
                {
                    sprintf(val, "%d", dev->new_val.a);
                }
                else if (dev->type == 3)
                {
                    sprintf(val, "%f", dev->new_val.b);
                }
                cJSON_AddItemToObject(item1, "val", cJSON_CreateString(val));
                cJSON_AddItemToArray(data1, item1);
                dev++;
            }

            publicmqtt(cJSON_Print(root1));
        }
    }
    while (1)
    {
        pause();
    }
    sqlite3_close(db);

    printf("Waiting for up to %d seconds for publication of %s\n"
           "on topic %s for client with ClientID: %s\n",
           (int)(TIMEOUT / 1000), PAYLOAD, TOPIC, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;

    return 0;
}
